package com.wdl.user

import org.apache.flink.api.common.state.{ListState, ListStateDescriptor}
import org.apache.flink.api.java.tuple.Tuple
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.functions.KeyedProcessFunction
import org.apache.flink.util.Collector

import scala.collection.mutable.ListBuffer

class TopNHotItems(size: Int) extends KeyedProcessFunction[Tuple, ItemViewCount, String]{

  override def open(parameters: Configuration): Unit = {

  }

  /** 定义一个 listState， 用来保存当前窗口所有的 count 结果 */
  lazy val itemViewCountState: ListState[ItemViewCount] =
    getRuntimeContext.getListState(new ListStateDescriptor[ItemViewCount]("", classOf[ItemViewCount]))
  override def processElement(value: ItemViewCount,
                              ctx: KeyedProcessFunction[Tuple, ItemViewCount, String]#Context,
                              out: Collector[String]): Unit = {
    /** 来一条数据，就把它保存到状态中 */
    itemViewCountState.add(value)

    /** 注册定时器，在 windowEnd + 1 触发 */
    ctx.timerService().registerEventTimeTimer( value.windowEnd + 1)

  }

  /** 定时器触发的时候，从状态中取出数据，然后排序输出 */
  override def onTimer(timestamp: Long,
                       ctx: KeyedProcessFunction[Tuple, ItemViewCount, String]#OnTimerContext,
                       out: Collector[String]): Unit = {
    /** 先把状态中的数据提取到一个 listBuffer 中 */
    val itemCountList: ListBuffer[ItemViewCount] = ListBuffer()
    import scala.collection.JavaConversions._
    for ( i <- itemViewCountState.get()){
      itemCountList += i
    }

    /** 按照 count 值的大小排序，取 topN */
    val sortedItemViewCounts: ListBuffer[ItemViewCount] =
      /** Ordering.Long.reverse 柯里化，按降序排列 */
      itemCountList.sortBy(_.count)(Ordering.Long.reverse).take(size)

    /** 清除状态 */
    itemViewCountState.clear()

    /** 将排名信息格式化成 String，方便监控 */
    val result: StringBuilder = new StringBuilder
    /** 遍历 sorted 表，输出 topN */
    for( i <- sortedItemViewCounts.indices ){
      /** 获取当前商品的 count 信息 */
      val currentItemCount = sortedItemViewCounts(i)
      result.append("Top").append(i+1).append(":")
        .append(" 商品ID=").append(currentItemCount.itemId)
        .append(" 访问量=").append(currentItemCount.count)
        .append("\n")
    }
    result.append("==============================\n\n")

    // 控制输出频率
    Thread.sleep(1000)
    out.collect(result.toString())
  }
}
